package com.ziroom.crm.mq.core.resend;

import com.ziroom.crm.mq.core.MqType;
import com.ziroom.crm.mq.core.config.MqConfig;
import com.ziroom.crm.mq.core.data.Message;
import com.ziroom.crm.mq.core.data.ProcessTypes;
import com.ziroom.crm.mq.core.logging.Log;
import com.ziroom.crm.mq.core.logging.LogFactory;
import com.ziroom.crm.mq.core.persistence.Persistence;
import com.ziroom.crm.mq.core.persistence.PersistenceManager;
import lombok.Getter;
import lombok.Setter;

import java.util.*;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * <p></p>
 * <p>
 * <PRE>
 * <BR>    修改记录
 * <BR>-----------------------------------------------
 * <BR>    修改日期         修改人          修改内容
 * </PRE>
 *
 * @author WeiShaoying
 * @version 1.0
 * @date Created in 2020年09月04日 10:29
 * @since 1.0
 */
@SuppressWarnings("ALL")
public class ReSendInvoker {

    private static final Log log = LogFactory.getLog(ReSendInvoker.class);

    /**
     * 一次获取个数
     */
    private static final int DEFAULT_SIZE = 500;

    private static List<ReSend> resendList;

    private static Map<MqType, ReSend> resendMap;

    private static Set<MqType> noConfig = new CopyOnWriteArraySet<>();

    @Getter
    @Setter
    public static Consumer<List<? extends Message>> callback;

    @Getter
    @Setter
    public static Consumer<Integer> countCallback;

    /**
     * 重新发送
     */
    public static void resend() {
        Map<MqType, ReSend> resendMap = getReSendMap();

        for (MqType mqType : MqType.values()) {
            if (noConfig.contains(mqType)) {
                continue;
            }
            ReSend resend = resendMap.get(mqType);
            if (Objects.isNull(resend)) {
                //对应的mq重试不存在
                addNoConfig(mqType);
                continue;
            }
            Persistence persistence = PersistenceManager.getPersistence(mqType);
            if (Objects.isNull(persistence)) {
                //对应的mq持久化不存在
                addNoConfig(mqType);
                continue;
            }
            //获取发送和重试失败的记录
            //当是文件的时候，先取持久化数据，内存失败数据可以充当一个缓冲队列使用
            List<? extends Message> dataList = ReSendInvoker.getNeedReSendMessage(persistence, ProcessTypes.RESEND, ProcessTypes.SEND);
            Integer threshold = MqConfig.getConfigInt(MqConfig.RESEND_THRESHOLD, DEFAULT_SIZE);
            if (dataList != null && dataList.size() > 0) {
                if (callback != null && dataList.size() >= threshold) {
                    callback.accept(dataList);
                }
                for (Message message : dataList) {
                    log.info("[rock-mq] Resend message id:{}", message.getId());
                    resend.reSend(message);
                }
            }
        }
    }

    public static List<? extends Message> getFaultMessage() {
        Persistence persistence = PersistenceManager.getPersistence(MqType.RABBIT);
        return persistence.queryAllMessage(ProcessTypes.DEAD_SEND, ProcessTypes.RESEND, ProcessTypes.SEND,
                ProcessTypes.SUCCESS_SEND);
    }

    public static int getRetryCount() {
        Persistence persistence = PersistenceManager.getPersistence(MqType.RABBIT);
        return persistence.getDataCount(ProcessTypes.RESEND, ProcessTypes.SEND);
    }

    public static void clear() {
        Persistence persistence = PersistenceManager.getPersistence(MqType.RABBIT);
        persistence.clear(ProcessTypes.DEAD_SEND, ProcessTypes.RESEND, ProcessTypes.SEND, ProcessTypes.SUCCESS_SEND);
    }

    public static int computeCount() {
        Persistence persistence = PersistenceManager.getPersistence(MqType.RABBIT);
        int dataCount = persistence.getDataCount(ProcessTypes.DEAD_SEND, ProcessTypes.RESEND, ProcessTypes.SEND,
                ProcessTypes.SUCCESS_SEND);
        if (countCallback != null) {
            countCallback.accept(dataCount);
        }
        return dataCount;
    }

    private static List<ReSend> loadReSendList() {
        // cache
        if (Objects.isNull(resendList)) {
            resendList = ReSendFactory.getReSendList();
        }
        return resendList;
    }

    private static Map<MqType, ReSend> getReSendMap() {
        if (Objects.isNull(resendMap)) {
            resendMap = loadReSendList().stream().collect(Collectors.toConcurrentMap(ReSend::getType, v -> v, (v1, v2) -> v1));
        }
        return resendMap;
    }

    private static void addNoConfig(MqType mqType) {
        noConfig.add(mqType);
    }

    /**
     * 返回所有需要发送的消息记录
     *
     * @param persistence
     * @param processType
     * @return
     */
    private static List<? extends Message> getNeedReSendMessage(Persistence persistence, ProcessTypes... processType) {
        List<? extends Message> result = new ArrayList<>();
        int needSize = ReSendInvoker.getSize();
        for (ProcessTypes type : processType) {
            List list = persistence.queryMessage(needSize, type, new Date());
            if (Objects.nonNull(list) && list.size() > 0) {
                result.addAll(list);
            }
            // 减少多次io，提升性能
            if (result.size() >= needSize) {
                break;
            }
        }
        return result;
    }

    /**
     * 获取批次数量
     *
     * @return 批次数量
     */
    private static int getSize() {
        return MqConfig.getConfigInt(MqConfig.RESEND_FETCH_SIZE, DEFAULT_SIZE);
    }

}
